/*
 *  Licensed to the Apache Software Foundation (ASF) under one
 *  or more contributor license agreements.  See the NOTICE file
 *  distributed with this work for additional information
 *  regarding copyright ownership.  The ASF licenses this file
 *  to you under the Apache License, Version 2.0 (the
 *  "License"); you may not use this file except in compliance
 *  with the License.  You may obtain a copy of the License at
 *
 *    http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing,
 *  software distributed under the License is distributed on an
 *  "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
 *  KIND, either express or implied.  See the License for the
 *  specific language governing permissions and limitations
 *  under the License.
 *
 */
package org.apache.asyncweb.server.transport.mina;

import java.io.IOException;
import java.net.InetSocketAddress;

import org.apache.asyncweb.common.DefaultHttpRequest;
import org.apache.asyncweb.common.DefaultHttpResponse;
import org.apache.asyncweb.common.HttpRequest;
import org.apache.asyncweb.common.HttpResponseStatus;
import org.apache.asyncweb.common.HttpVersion;
import org.apache.asyncweb.common.MutableHttpResponse;
import org.apache.asyncweb.common.codec.HttpCodecFactory;
import org.apache.asyncweb.common.codec.HttpRequestDecoderException;
import org.apache.asyncweb.server.HttpServiceContext;
import org.apache.asyncweb.server.HttpServiceFilter;
import org.apache.asyncweb.server.ServiceContainer;
import org.apache.asyncweb.server.context.AbstractHttpServiceContext;
import org.apache.asyncweb.server.pipeline.RequestPipeline;
import org.apache.asyncweb.server.pipeline.RequestPipelineListener;
import org.apache.mina.core.filterchain.IoFilterAdapter;
import org.apache.mina.core.future.IoFutureListener;
import org.apache.mina.core.future.WriteFuture;
import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.core.write.DefaultWriteRequest;
import org.apache.mina.core.write.WriteRequest;
import org.apache.mina.filter.codec.ProtocolCodecFilter;
import org.apache.mina.filter.codec.ProtocolDecoderException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * The default HttpIoHandler used when one is not provided.
 * 
 * @author The Apache MINA Project (dev@mina.apache.org)
 */
public class DefaultHttpIoHandler extends IoHandlerAdapter {
	private static final Logger LOG = LoggerFactory
			.getLogger(DefaultHttpIoHandler.class);
	
	public static final String WRITE_FUTURE = "WRITE_FUTURE";
	
	public static final String SERVICE_CONTAINER = "SERVICE_CONTAINER";
	
	public static final String HTTP_SERVICE_CONTEXT = "HTTP_SERVICE_CONTEXT";
	
	/** the default idle time in seconds - 5 minutes */
	public static final int DEFAULT_IDLE_TIME = 300;

	/** the HttpService container **/
	private ServiceContainer container;

	/** the request pipeline */
	private RequestPipeline pipeline;

	/** idle time for request reads */
	private int readIdleTime = DEFAULT_IDLE_TIME;

	public DefaultHttpIoHandler(ServiceContainer container,
			RequestPipeline pipeline) {
		this(container, pipeline, DEFAULT_IDLE_TIME);
	}

	public DefaultHttpIoHandler(ServiceContainer container,
			RequestPipeline pipeline, int readIdleTime) {
		super();
		this.container = container;
		this.pipeline = pipeline;
		if (readIdleTime > 0)
			this.readIdleTime = readIdleTime;
	}

	@Override
	public void sessionCreated(IoSession session) throws Exception {
		session.setAttribute(SERVICE_CONTAINER, container);
		session.getConfig().setIdleTime(IdleStatus.READER_IDLE, readIdleTime);
		session.getFilterChain().addLast("codec",
				new ProtocolCodecFilter(new HttpCodecFactory()));
		session.getFilterChain().addLast("converter",
				new ServiceContextFilter());
		session.getFilterChain().addLast("pipeline",
				new RequestPipelineAdapter(pipeline));

		int i = 0;
		for (HttpServiceFilter serviceFilter : container.getServiceFilters()) {
			session.getFilterChain().addLast("serviceFilter." + i++,
					new ServiceFilterAdapter(serviceFilter));
		}
		if (LOG.isDebugEnabled())
			LOG.debug("Session created: {}", session.getRemoteAddress());
	}

	@Override
	public void sessionOpened(IoSession session) throws Exception {
		if (LOG.isDebugEnabled())
			LOG.debug("Session opened: {}", session.getRemoteAddress());
	}

	@Override
	public void sessionClosed(IoSession session) throws Exception {
		if (LOG.isDebugEnabled())
			LOG.debug("Session closed: {}", session.getRemoteAddress());
		HttpServiceContext serviceContext = (HttpServiceContext) session
				.getAttribute(HTTP_SERVICE_CONTEXT);
		if (serviceContext != null) {
			serviceContext.fireClientDisconnected();
		}
		session.removeAttribute(HTTP_SERVICE_CONTEXT);
		session.removeAttribute(SERVICE_CONTAINER);
	}

	@Override
	public void sessionIdle(IoSession session, IdleStatus status)
			throws Exception {
		if (session.getIdleCount(status) >= 1) {
			HttpServiceContext context = (HttpServiceContext) session
					.getAttribute(HTTP_SERVICE_CONTEXT);
			if (LOG.isDebugEnabled() && context != null)
				LOG.debug(
						"Session idle detected on context {} with idleType {}",
						context, status);

			if (context != null) {
				if (IdleStatus.BOTH_IDLE == status
						|| IdleStatus.READER_IDLE == status) {
					context.fireClientIdle(session.getLastReaderIdleTime(),
							session.getReaderIdleCount());
				}
			} else {
				// TODO - look further into this - it may present serious issues
				// when dealing with HTTP/1.1
				if (LOG.isDebugEnabled())
					LOG.debug("Idled with no current request. Scheduling closure when pipeline empties");
				final IoSession ioSession = session;
				pipeline.runWhenEmpty(new Runnable() {
					public void run() {
						if (LOG.isDebugEnabled())
							LOG.debug(
									"Pipeline empty after idle. Closing session: {}",
									ioSession.getRemoteAddress());
						ioSession.close(false);
					}
				});
			}
		}
	}

	@Override
	public void exceptionCaught(IoSession session, Throwable cause)
			throws Exception {
		MutableHttpResponse response = null;

		if (cause instanceof ProtocolDecoderException) {
			HttpResponseStatus status;

			if (cause instanceof HttpRequestDecoderException) {
				status = ((HttpRequestDecoderException) cause)
						.getResponseStatus();
			} else {
				status = HttpResponseStatus.BAD_REQUEST;
			}
			if (LOG.isWarnEnabled())
				LOG.warn("Bad request: {}", session.getRemoteAddress(), cause);

			response = new DefaultHttpResponse();
			response.setProtocolVersion(HttpVersion.HTTP_1_1);
			response.setStatus(status);
		} else if (cause instanceof IOException) {
			if (LOG.isWarnEnabled())
				LOG.warn("IOException on HTTP connection", cause);
			session.close(false);
		} else {
			response = new DefaultHttpResponse();
			response.setProtocolVersion(HttpVersion.HTTP_1_1);
			response.setStatus(HttpResponseStatus.INTERNAL_SERVER_ERROR);
			if (LOG.isErrorEnabled())
				LOG.error(
						"Unexpected exception from a service : "
								+ session.getRemoteAddress(), cause);
		}

		if (response != null) {
			HttpServiceContext context = (HttpServiceContext) session
					.getAttribute(HTTP_SERVICE_CONTEXT);
			if (context == null) {
				context = new DefaultHttpServiceContext(session,
						new DefaultHttpRequest(), container);
			}
			context.commitResponse(response);
		}
	}

	private class DefaultHttpServiceContext extends AbstractHttpServiceContext {
		private IoSession ioSession;

		public DefaultHttpServiceContext(IoSession session,
				HttpRequest request, ServiceContainer container) {
			super((InetSocketAddress) session.getLocalAddress(),
					(InetSocketAddress) session.getRemoteAddress(), request,
					container);
			this.ioSession = session;
			session.setAttribute(HTTP_SERVICE_CONTEXT, this);
		}

		@Override
		protected void doWrite(boolean requiresClosure) {
			ioSession.removeAttribute(HTTP_SERVICE_CONTEXT);
			WriteFuture future = ioSession.write(this);
			if (requiresClosure) {
				if (LOG.isDebugEnabled())
					LOG.debug("Added CLOSE future listener.");
				future.addListener(IoFutureListener.CLOSE);
			}
		}

	}

	private class ServiceContextFilter extends IoFilterAdapter {

		@Override
		public void filterWrite(NextFilter nextFilter, IoSession session,
				WriteRequest writeRequest) throws Exception {
			HttpServiceContext serviceContext = (HttpServiceContext) writeRequest
					.getMessage();
			nextFilter.filterWrite(
					session,
					new DefaultWriteRequest(serviceContext
							.getCommittedResponse(), writeRequest.getFuture()));
		}

		@Override
		public void messageReceived(NextFilter nextFilter, IoSession session,
				Object message) throws Exception {
			ServiceContainer container = (ServiceContainer) session
					.getAttribute(SERVICE_CONTAINER);
			HttpRequest request = (HttpRequest) message;
			HttpServiceContext serviceContext = new DefaultHttpServiceContext(
					session, request, container);
			nextFilter.messageReceived(session, serviceContext);
		}
	}

	private class ServiceFilterAdapter extends IoFilterAdapter {
		private final HttpServiceFilter filter;

		public ServiceFilterAdapter(HttpServiceFilter filter) {
			this.filter = filter;
		}

		@Override
		public void messageReceived(final NextFilter nextFilter,
				final IoSession session, final Object message) throws Exception {
			HttpServiceFilter.NextFilter nextFilterAdapter = new HttpServiceFilter.NextFilter() {
				public void invoke() {
					nextFilter.messageReceived(session, message);
				}
			};
			filter.handleRequest(nextFilterAdapter,
					(HttpServiceContext) message);
		}

		@Override
		public void filterWrite(final NextFilter nextFilter,
				final IoSession session, final WriteRequest writeRequest)
				throws Exception {
			HttpServiceFilter.NextFilter nextFilterAdapter = new HttpServiceFilter.NextFilter() {
				public void invoke() {
					nextFilter.filterWrite(session, writeRequest);
				}
			};

			HttpServiceContext context = (HttpServiceContext) writeRequest
					.getMessage();
			filter.handleResponse(nextFilterAdapter, context);
		}
	}

	private class RequestPipelineAdapter extends IoFilterAdapter {

		private final RequestPipeline pipeline;

		public RequestPipelineAdapter(RequestPipeline pipeline) {
			this.pipeline = pipeline;
		}

		@Override
		public void sessionOpened(final NextFilter nextFilter,
				final IoSession session) {
			pipeline.setPipelineListener(new RequestPipelineListener() {
				public void responseReleased(HttpServiceContext context) {
					WriteFuture future=(WriteFuture)session.getAttribute(WRITE_FUTURE);
					nextFilter.filterWrite(session, new DefaultWriteRequest(
							context, future));
				}
			});

			nextFilter.sessionOpened(session);
		}

		@Override
		public void messageReceived(NextFilter nextFilter, IoSession session,
				Object message) throws Exception {
			HttpServiceContext context = (HttpServiceContext) message;
			if (pipeline.addRequest(context)) {
				if (LOG.isDebugEnabled())
					LOG.debug("Allocated slot in request pipeline");
				nextFilter.messageReceived(session, message);
			} else {
				// The client has filled their pipeline. Currently, this
				// triggers closure. Another option would be to drop read
				// interest until we drain.
				if (LOG.isWarnEnabled())
					LOG.warn("Could not allocate room in the pipeline for request");
				MutableHttpResponse response = new DefaultHttpResponse();
				response.setStatusReasonPhrase("Pipeline full");
				response.setStatus(HttpResponseStatus.SERVICE_UNAVAILABLE);
				context.commitResponse(response);
			}
		}

		@Override
		public void filterWrite(NextFilter nextFilter, IoSession session,
				WriteRequest writeRequest) throws Exception {
			HttpServiceContext context = (HttpServiceContext) writeRequest
					.getMessage();
			session.setAttribute(WRITE_FUTURE,writeRequest.getFuture());
			pipeline.releaseResponse(context);
			// nextFilter will be invoked when pipeline listener is notified.
		}
	}
}
